package io.plus.interactive.task;

import drds.common.Clock;
import io.plus.interactive.AsynchronousSocketChannelContext;
import io.plus.interactive.Configuration;
import io.plus.interactive.PacketHandlerMode;
import io.plus.interactive.user_interface.Packet;
import io.plus.utils.lock.ReadWriteLockMap;
import io.plus.utils.queue.Queue;
import io.plus.utils.queue.QueueImpl;
import io.plus.utils.thread.lock_capable_task.AbstractQueueTask;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.Executor;
import java.util.concurrent.atomic.AtomicLong;

@Slf4j
public class HandlerTask extends AbstractQueueTask<Packet> {
    //private static final Logger log = LoggerFactory.getLogger(HandlerTask.class);

    private AsynchronousSocketChannelContext asynchronousSocketChannelContext = null;
    private Configuration configuration = null;

    private AtomicLong synFailCount = new AtomicLong();
    private Queue<Packet> queue = null;

    public HandlerTask(AsynchronousSocketChannelContext asynchronousSocketChannelContext, Executor executor) {
        super(executor);
        this.asynchronousSocketChannelContext = asynchronousSocketChannelContext;
        configuration = asynchronousSocketChannelContext.clientConfiguration;
        getQueue();
    }

    /**
     * 处理packet
     */
    public void handler(Packet packet) {
        long start = Clock.currentTimeMillis;
        try {
            Integer synSeq = packet.getSynSeq();
            if (synSeq != null && synSeq > 0) {
                ReadWriteLockMap<Integer, Packet> syns = configuration.getWaitingResps();
                Packet initPacket = syns.remove(synSeq);
                if (initPacket != null) {
                    synchronized (initPacket) {
                        syns.put(synSeq, packet);
                        initPacket.notify();
                    }
                } else {
                    log.error("[{}]同步消息失败, synSeq is {}, 但是同步集合中没有对应key值", synFailCount.incrementAndGet(), synSeq);
                }
            } else {
                configuration.getAioHandler().handler(asynchronousSocketChannelContext, packet);
            }
        } catch (Throwable e) {
            log.error(packet.toString(), e);
        } finally {
            long end = Clock.currentTimeMillis;
            long iv = end - start;


            if (configuration.getAioListener() != null) {
                try {
                    configuration.getAioListener().onAfterHandled(asynchronousSocketChannelContext, packet, iv);
                } catch (Exception e) {
                    log.error(e.toString(), e);
                }
            }

        }
    }


    @Override
    public void runTask() {
        Packet packet = null;
        while ((packet = queue.poll()) != null) {
            handler(packet);
        }
    }

    @Override
    public String toString() {
        return this.getClass().getSimpleName() + ":" + asynchronousSocketChannelContext.toString();
    }


    @Override
    public Queue<Packet> getQueue() {
        if (PacketHandlerMode.QUEUE == configuration.packetHandlerMode) {
            if (queue == null) {
                synchronized (this) {
                    if (queue == null) {
                        queue = new QueueImpl<Packet>();
                    }
                }
            }
            return queue;
        }
        return null;
    }

}
